package com.alibaba.cirtus.ots.application.adapter.ability.facade.serviceV2.sap;

import com.alibaba.cirtus.ots.application.adapter.ability.model.enums.MnSAPStatus;
import com.alibaba.cirtus.ots.application.adapter.ability.model.enums.PriceEnum;
import com.alibaba.cirtus.ots.application.adapter.ability.model.request.PriceTableDataSapRequest;
import com.alibaba.cirtus.ots.application.adapter.ability.model.request.PriceTableKonpDataSapRequest;
import com.alibaba.cirtus.ots.application.adapter.ability.model.response.CallApiRecordResponse;
import com.alibaba.cirtus.ots.application.adapter.ability.model.response.PriceDimensionResponse;
import com.alibaba.cirtus.ots.application.adapter.ability.utils.PageRecordUtil;
import com.alibaba.citrus.ots.common.annotation.ProcessInvoker;
import com.alibaba.citrus.ots.common.util.IntegrateUtils;
import com.alibaba.citrus.ots.common.util.SapAndCrmFieldDealUtils;
import com.alibaba.cz.base.tool.Nullable;
import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.lattice2.epoch.sdk.model.Result;
import com.epoch.app.crpricecenter.dto.request.DimensionProperty;
import com.epoch.app.crpricecenter.dto.request.SyncPriceDimensionConfigDataRequest;
import com.epoch.app.crpricecenter.model.dto.PriceDimensionRequest;
import com.epoch.app.crpricecenter.model.dto.SyncPricePropertyConfigDataResponse;
import com.epoch.app.crpricecenter.model.dto.SyncPriceRecordDataRequest;
import com.epoch.app.crpricecenter.model.dto.SyncPriceRecordDataResponse;
import com.epoch.app.crpricecenter.sdo.PriceDimensionSDO;
import com.epoch.app.crpricecenter.service.CrPriceCenterService;
import com.epoch.app.otsapplicationadaptersdk.contants.OtsApplicationAdapterSdkConstants;
import com.epoch.app.otsapplicationadaptersdk.model.dto.PriceSyncTempDimensionLoadListVO;
import com.epoch.app.otsapplicationadaptersdk.model.dto.PriceSyncTempLoadListVO;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.BatchCreateRequest;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.BatchDeleteRequest;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.BatchDeleteResponse;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.GroupByMessageIdRequest;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.UpdateRequest;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.dto.UpdateResponse;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.model.PriceSyncTemp;
import com.epoch.app.otsapplicationadaptersdk.pricesynctemp.service.PriceSyncTempService;
import com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.model.PriceSyncTempDimension;
import com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.service.PriceSyncTempDimensionService;
import com.epoch.app.otsapplicationadaptersdk.sap.MnSapCommResponse;
import com.epoch.app.otsapplicationadaptersdk.sap.MnSapCommSDO;
import com.epoch.app.otsapplicationadaptersdk.sapv2.mnsapif001v2.dto.DoHandlerTempTableRequest;
import com.epoch.app.otsapplicationadaptersdk.sapv2.mnsapif001v2.dto.ExternalSyncPriceRequest;
import com.epoch.app.otsapplicationadaptersdk.sapv2.mnsapif001v2.service.MnSapIf001V2Service;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * @author hanguo
 * @Description sap请求平台同步价格
 * @date 2021/9/25
 */
@Component
@Primary
public class MnSapIf001V2ServiceImpl implements MnSapIf001V2Service {
    private final Log log = Log.getLogger(MnSapIf001V2ServiceImpl.class);

    private static final String MESSAGE_ID = "messageId";

    private static final String DATA_SOURCE = "sender";

    @Resource
    private CrPriceCenterService crPriceCenterService;

    @Resource
    private PriceSyncTempService priceSyncTempService;

    @Resource
    private PriceSyncTempDimensionService priceSyncTempDimensionService;

    @Resource(name="scItemsQueryTaskExecutor")
    ThreadPoolTaskExecutor scItemsQueryTaskExecutor;

    @Override
    @ProcessInvoker(remark = "MnSapIf001V2Service", errorCode = "OTS-02-007-00-16-999", printLog = true)
    public MnSapCommResponse externalSyncPrice(ExternalSyncPriceRequest externalSyncPriceRequest) {
        //messageHeader
        Map<String, Object> messageHeader = Maps.newHashMap();
        if (Objects.nonNull(externalSyncPriceRequest.getMessageHeader())
                && MapUtils.isNotEmpty((Map<String, Object>)externalSyncPriceRequest.getMessageHeader())){
            messageHeader = (Map<String, Object>) externalSyncPriceRequest.getMessageHeader();
        }else {
            Map<String,Object> orginalMessageHeader = IntegrateUtils.safeGetHeaderMap();
            if (MapUtils.isNotEmpty(orginalMessageHeader)){
                messageHeader =  orginalMessageHeader;
            }
        }
        //data
        Map<String, Object> data = Maps.newHashMap();
        if (Objects.nonNull(externalSyncPriceRequest.getData())
                && MapUtils.isNotEmpty((Map<String, Object>)externalSyncPriceRequest.getData())){
            data = (Map<String, Object>) externalSyncPriceRequest.getData();
        }else {
            Object orginalData = IntegrateUtils.safeGetDataObject();
            if (Objects.nonNull(orginalData)){
                data = (Map<String, Object>) orginalData;
            }
        }

        MnSapCommResponse mnSapCommResponse = new MnSapCommResponse();
        mnSapCommResponse.setInterfacePath((String) messageHeader.get("interface"));
        mnSapCommResponse.setMessageId((String) messageHeader.get("messageId"));

        if (Objects.isNull(data)) {
            return mnSapCommResponse;
        }

        //不接收超过2M的数据
//        int length = JSONObject.toJSONString(externalSyncPriceRequest).getBytes().length;
//        if(length > PriceEnum.REQUEST_MESSAGE_BYTE_SIZE.getValue()){
//                mnSapCommSDO.setStatus(MnSAPStatus.ERROR.getCode());
//                mnSapCommSDO.setMessage("数据量过大,不能超过"+PriceEnum.REQUEST_MESSAGE_BYTE_SIZE.getValue()+"字节，当前数据大小为"+length+"字节");
//                mnSapCommSDOS.add(mnSapCommSDO);
//                mnSapCommResponse.setData(mnSapCommSDOS);
//                return mnSapCommResponse;
//        }

        List<MnSapCommSDO> mnSapCommSDOS = Lists.newArrayList();
        MnSapCommSDO mnSapCommSDO = new MnSapCommSDO();

        //20220817 二期分子公司上线，处理大批量数据，走异步处理,不要同步es
        if("1".equals(OtsApplicationAdapterSdkConstants.sap_sync_price_mode())){
            String messageId = (String) messageHeader.get("messageId");
            if(StringUtils.isBlank(messageId)){
                mnSapCommSDO.setStatus(MnSAPStatus.ERROR.getCode());
                mnSapCommSDO.setMessage("messageId不能为空");
                mnSapCommSDOS.add(mnSapCommSDO);
                mnSapCommResponse.setData(mnSapCommSDOS);
                return mnSapCommResponse;
            }
            String sendTime = (String) messageHeader.get("sendTime");
            String sender = (String) messageHeader.get("sender");

            List<Future<Long>> futureList = Lists.newArrayList();
            CompletableFuture<Long> task1 = saveKnopPriceSyncTempTable(data,messageId,sendTime,sender);
            futureList.add(task1);
            CompletableFuture<Long> task2 = saveDimensionPriceSyncTempTable(data,messageId,sendTime,sender);
            futureList.add(task2);

            long count = 0;
            try {
                for (Future<Long> future : futureList){
                    Long subList = future.get();
                    count += subList;
                }
            }catch (Exception e){
                log.error("MnSapIf001V2Service_future_get_error={}", e.getMessage());
            }

            mnSapCommSDO.setDref1(count);
            mnSapCommSDO.setStatus(MnSAPStatus.SUCCESS.getCode());
            mnSapCommSDO.setMessage("同步成功");
        }else{
            //操作过的表结果记录
            List<CallApiRecordResponse> callApiRecordResponses = Lists.newArrayList();

            long konpCount = extractedPriceRecord(data, mnSapCommSDO, callApiRecordResponses,messageHeader);

            Map<String, PriceDimensionResponse> dimensionMap = queryPriceDimensionResponseMap(data);

            long tableCount = savePricePropertyConfig(data, mnSapCommSDO, callApiRecordResponses, dimensionMap,messageHeader);
            log.info("MnSapIf001V2Service_externalSyncPrice_receive_tableCounts_total_size={}", tableCount);
            log.info("MnSapIf001V2Service_externalSyncPrice_sync_result={}", JSONObject.toJSONString(callApiRecordResponses));

            //记录收到的数据条数
            mnSapCommSDO.setDref1(konpCount + tableCount);
            //没有失败记录时表示同步数据成功
            if (StringUtils.isBlank(mnSapCommSDO.getStatus()) && StringUtils.isBlank(mnSapCommSDO.getMessage())) {
                mnSapCommSDO.setStatus(MnSAPStatus.SUCCESS.getCode());
                mnSapCommSDO.setMessage("同步成功");
            }
        }

        mnSapCommSDOS.add(mnSapCommSDO);
        mnSapCommResponse.setData(mnSapCommSDOS);
        return mnSapCommResponse;
    }



    /**
     * 保存价格属性配置
     *
     * @param data
     * @param mnSapCommSDO
     * @param callApiRecordResponses
     * @param dimensionMap
     */
    private long savePricePropertyConfig(Map<String, Object> data, MnSapCommSDO mnSapCommSDO,
                                         List<CallApiRecordResponse> callApiRecordResponses,
                                         Map<String, PriceDimensionResponse> dimensionMap,
                                         Map<String,Object> messageHeader) {

        long count = 0;
        //同步价格维度数据，读完一张表的数据后写入,逐行写数据
        List<String> tableDataList = data.keySet().stream().filter(key -> !StringUtils.equals(key, "KONP")).collect(Collectors.toList());
        log.info("MnSapIf001V2Service_externalSyncPrice_receive_data_tableDataList={}",tableDataList);
        for (String tableName : tableDataList) {
            //获取维度id和租户id
            String dimensionId = dimensionMap.get(tableName).getDimensionId();
            String tenantId = dimensionMap.get(tableName).getTenantId();

            List<Object> table = (List<Object>) data.get(tableName);
            log.info("MnSapIf001V2Service_externalSyncPrice_receive_data_tableName,forEach_tableName={},receive_size={}",tableName,table.size());
            if (CollectionUtils.isEmpty(table)) {
                continue;
            }
            count += table.size();

            List<PriceTableDataSapRequest> priceTableDataSapRequests = JSONObject.parseArray(JSONObject.toJSONString(table), PriceTableDataSapRequest.class);
            if (CollectionUtils.isEmpty(priceTableDataSapRequests)) {
                continue;
            }

            int tableSuccessNum = 0;
            for(PriceTableDataSapRequest tableValue : priceTableDataSapRequests) {
                SyncPricePropertyConfigDataResponse syncPricePropertyConfigDataResponse = new SyncPricePropertyConfigDataResponse();
                SyncPriceDimensionConfigDataRequest syncPriceDimensionConfigDataRequest = new SyncPriceDimensionConfigDataRequest();

                CallApiRecordResponse callAPISyncPricePropertyConfigDataResponse = new CallApiRecordResponse();
                try {
                    syncPriceDimensionConfigDataRequest.setTableName(tableName);

                    syncPriceDimensionConfigDataRequest.setDimensionId(dimensionId);

                    List<DimensionProperty> dimensionPropertyList = Lists.newArrayList();
                    {
                        //填入维度组属性数据
                        DimensionProperty orgSalesOrganizationCodeSDO = new DimensionProperty();
                        orgSalesOrganizationCodeSDO.setDimensionName("org_sales_organization_code");
                        orgSalesOrganizationCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getOrgSalesOrganizationCode()));
                        dimensionPropertyList.add(orgSalesOrganizationCodeSDO);

                        DimensionProperty channelCodeSDO = new DimensionProperty();
                        channelCodeSDO.setDimensionName("channel_code");
                        channelCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getChannelCode()));
                        dimensionPropertyList.add(channelCodeSDO);

                        DimensionProperty scItemGroupCodeSDO = new DimensionProperty();
                        scItemGroupCodeSDO.setDimensionName("sc_item_group_code");
                        scItemGroupCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getScItemGroupCode()));
                        dimensionPropertyList.add(scItemGroupCodeSDO);

                        DimensionProperty customerAddressCodeSDO = new DimensionProperty();
                        customerAddressCodeSDO.setDimensionName("customer_send_address_code");
                        customerAddressCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getCustomerAddressCode()));
                        dimensionPropertyList.add(customerAddressCodeSDO);

                        DimensionProperty salesUnitCodeSDO = new DimensionProperty();
                        salesUnitCodeSDO.setDimensionName("sales_unit_code");
                        salesUnitCodeSDO.setDimensionValue(tableValue.getSalesUnitCode());
                        dimensionPropertyList.add(salesUnitCodeSDO);

                        DimensionProperty scItemCodeSDO = new DimensionProperty();
                        scItemCodeSDO.setDimensionName("material_code");
                        scItemCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getMaterialCode()));
                        dimensionPropertyList.add(scItemCodeSDO);

                        DimensionProperty customerSaleAddressCodeSDO = new DimensionProperty();
                        customerSaleAddressCodeSDO.setDimensionName("customer_sale_address_code");
                        customerSaleAddressCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getCustomerCode()));
                        dimensionPropertyList.add(customerSaleAddressCodeSDO);

                        DimensionProperty salesAreaSDO = new DimensionProperty();
                        salesAreaSDO.setDimensionName("sales_area");
                        salesAreaSDO.setDimensionValue(StringUtils.trim(tableValue.getSalesArea()));
                        dimensionPropertyList.add(salesAreaSDO);

                        DimensionProperty customerGroupSDO = new DimensionProperty();
                        customerGroupSDO.setDimensionName("customer_group");
                        customerGroupSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getCustomerGroup()));
                        dimensionPropertyList.add(customerGroupSDO);

                        DimensionProperty factoryCode = new DimensionProperty();
                        factoryCode.setDimensionName("factory_code");
                        factoryCode.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getFactoryCode()));
                        dimensionPropertyList.add(factoryCode);

                        DimensionProperty priceListCodeSDO = new DimensionProperty();
                        priceListCodeSDO.setDimensionName("price_list_code");
                        priceListCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getPriceListCode()));
                        dimensionPropertyList.add(priceListCodeSDO);

                        DimensionProperty storePlaceSDO = new DimensionProperty();
                        storePlaceSDO.setDimensionName("store_place");
                        storePlaceSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getInventoryCode()));
                        dimensionPropertyList.add(storePlaceSDO);

                        DimensionProperty salesDepartmentCodeSDO = new DimensionProperty();
                        salesDepartmentCodeSDO.setDimensionName("sales_department_code");
                        salesDepartmentCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getSaleDepartmentCode()));
                        dimensionPropertyList.add(salesDepartmentCodeSDO);

                        DimensionProperty salesGroupCodeSDO = new DimensionProperty();
                        salesGroupCodeSDO.setDimensionName("sales_group_code");
                        salesGroupCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getSaleDepartmentCode()));
                        dimensionPropertyList.add(salesGroupCodeSDO);

                        DimensionProperty customerCodeSDO = new DimensionProperty();
                        customerCodeSDO.setDimensionName("customer_code");
                        customerCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getCustomerCode()));
                        dimensionPropertyList.add(customerCodeSDO);


                        DimensionProperty departmentCodeSDO = new DimensionProperty();
                        departmentCodeSDO.setDimensionName("department_code");
                        departmentCodeSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getScItemGroupCode()));
                        dimensionPropertyList.add(departmentCodeSDO);

                        DimensionProperty priceAreaSDO = new DimensionProperty();
                        priceAreaSDO.setDimensionName("price_area");
                        priceAreaSDO.setDimensionValue(SapAndCrmFieldDealUtils.generateIdField(tableValue.getSalesArea()));
                        dimensionPropertyList.add(priceAreaSDO);

                    }
                    syncPriceDimensionConfigDataRequest.setDimensionPropertyList(dimensionPropertyList);

                    syncPriceDimensionConfigDataRequest.setPriceTypeId(tableValue.getPriceTypeId());
                    syncPriceDimensionConfigDataRequest.setPriceStartDate(SapAndCrmFieldDealUtils.forPriceStartDate(tableValue.getPriceStartDate()));
                    syncPriceDimensionConfigDataRequest.setPriceEndDate(SapAndCrmFieldDealUtils.forPriceEndDate(tableValue.getPriceEndDate()));
                    syncPriceDimensionConfigDataRequest.setPriceRecordId(tableValue.getPriceRecordId());
                    syncPriceDimensionConfigDataRequest.setTenantId(tenantId);
                    syncPriceDimensionConfigDataRequest.setMessageId((String) messageHeader.get(MESSAGE_ID));
                    syncPriceDimensionConfigDataRequest.setDataSource((String) messageHeader.get(DATA_SOURCE));

                    //调用api写入数据
                    syncPricePropertyConfigDataResponse = crPriceCenterService.syncPricePropertyConfigData(syncPriceDimensionConfigDataRequest);
                    log.info("MnSapIf001V2Service_externalSyncPrice_save_price_property_config_request={},result={}", syncPriceDimensionConfigDataRequest.toString(), JSONObject.toJSONString(syncPricePropertyConfigDataResponse));
                    if (syncPricePropertyConfigDataResponse.getSuccess()) {
                        tableSuccessNum++;
                    }
                    callAPISyncPricePropertyConfigDataResponse.setOperateResult(syncPricePropertyConfigDataResponse.getSuccess());
                    callAPISyncPricePropertyConfigDataResponse.setRequestBody(JSONObject.toJSONString(syncPriceDimensionConfigDataRequest));
                    callAPISyncPricePropertyConfigDataResponse.setTableName("ofs_price_property_config");
                    callAPISyncPricePropertyConfigDataResponse.setMessage(syncPricePropertyConfigDataResponse.getMessage());
                    callApiRecordResponses.add(callAPISyncPricePropertyConfigDataResponse);
                } catch (Exception e) {
                    mnSapCommSDO.setStatus(MnSAPStatus.ERROR.getCode());
                    mnSapCommSDO.setMessage(e.getMessage());

                    //写入结果数据
                    callAPISyncPricePropertyConfigDataResponse.setOperateResult(false);
                    callAPISyncPricePropertyConfigDataResponse.setRequestBody(JSONObject.toJSONString(syncPriceDimensionConfigDataRequest));
                    callAPISyncPricePropertyConfigDataResponse.setTableName("ofs_price_property_config");
                    callAPISyncPricePropertyConfigDataResponse.setMessage(e.getMessage());
                    callApiRecordResponses.add(callAPISyncPricePropertyConfigDataResponse);

                    log.error("MnSapIf001V2Service_externalSyncPrice_save_price_property_config_error={}", e.getMessage());
                }
            }
            log.info("MnSapIf001V2Service_externalSyncPrice_receive_data_tableName={},sync_success_size={}",tableName,tableSuccessNum);
            tableSuccessNum = 0;
        }

        return count;
    }

    @NotNull
    private Map<String, PriceDimensionResponse> queryPriceDimensionResponseMap(Map<String, Object> data) {
        List<String> tableNames = data.keySet().stream().filter(key -> !StringUtils.equals(key, "KONP")).collect(Collectors.toList());
        //获取表名称和维度的对应关系
        PriceDimensionRequest batchPriceDimensionRequest = new PriceDimensionRequest();
        batchPriceDimensionRequest.setTableNames(tableNames);
        Result<List<PriceDimensionSDO>> batchPriceDimensionResponse = crPriceCenterService.batchPriceDimension(batchPriceDimensionRequest);
        log.info("MnSapIf001V2Service_externalSyncPrice_get_table_dimension_relation_result = {}", JSONObject.toJSONString(batchPriceDimensionResponse));
        Map<String, PriceDimensionResponse> dimensionMap = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(batchPriceDimensionResponse.getResult())) {
            dimensionMap = JSONObject.parseArray(JSONObject.toJSONString(batchPriceDimensionResponse.getResult()), PriceDimensionResponse.class).stream().collect(Collectors.toMap(PriceDimensionResponse::getDimensionTableName, item -> item));
        }
        return dimensionMap;
    }

    private long extractedPriceRecord(Map<String, Object> data, MnSapCommSDO mnSapCommSDO,
                                      List<CallApiRecordResponse> callApiRecordResponses,
                                      Map<String,Object> messageHeader) {

        long count = 0;

        //同步价格记录数据
        Object konp = data.get("KONP");
        if (Objects.isNull(konp)) {
            return count;
        }

        List<PriceTableKonpDataSapRequest> priceTableKonpDataSapRequests = JSONObject.parseArray(JSONObject.toJSONString(konp), PriceTableKonpDataSapRequest.class);
        log.info("MnSapIf001V2Service_externalSyncPrice_receive_data_tableName=knop,receive_size={}",priceTableKonpDataSapRequests.size());
        if (CollectionUtils.isEmpty(priceTableKonpDataSapRequests)) {
            return count;
        }
        count += priceTableKonpDataSapRequests.size();


        int successNum = 0;
        for(PriceTableKonpDataSapRequest item : priceTableKonpDataSapRequests) {
            SyncPriceRecordDataRequest syncPriceRecordDataRequest = new SyncPriceRecordDataRequest();
            syncPriceRecordDataRequest.setPriceRecordId(item.getPriceRecordId());
            String sAmount = item.getAmount();
            if (StringUtils.isNotBlank(sAmount)) {

                BigDecimal amount = new BigDecimal(sAmount);
                BigDecimal multiply = new BigDecimal(PriceEnum.AMOUNT.getValue());
                syncPriceRecordDataRequest.setMoneyAmount(amount.multiply(multiply).longValue());
            }

            if (StringUtils.isNotBlank(item.getDeleteSignal()) && StringUtils.equals(PriceEnum.DELETE.getCode(), item.getDeleteSignal())) {
                    syncPriceRecordDataRequest.setStatus(PriceEnum.DELETE.getValue());
            }
            //分子
            String numerator = item.getNumerator();
            //分母
            String denominator = item.getDenominator();
            Long invalidData = -000000000L;
            //先写死
            syncPriceRecordDataRequest.setMoneyPercentage(invalidData.toString());
            syncPriceRecordDataRequest.setCurrency(item.getCurrency());
            syncPriceRecordDataRequest.setItemCount(SapAndCrmFieldDealUtils.toLong(item.getPriceUint()));
            syncPriceRecordDataRequest.setPriceUnite(item.getMeterUnit());
            syncPriceRecordDataRequest.setTenantId(item.getCustomerCode());
            syncPriceRecordDataRequest.setMessageId((String) messageHeader.get(MESSAGE_ID));
            syncPriceRecordDataRequest.setDataSource((String) messageHeader.get(DATA_SOURCE));

            //TODO need fix 这里的价格更新策略存在老数据覆盖旧数据问题，但是因为1. 价格变更频率不高，2. 页面有手工更新价格入口补偿； 可上线后修复

            //调用api写入数据, 同步价格技术数据
            SyncPriceRecordDataResponse syncPriceRecordDataResponse = crPriceCenterService.syncPriceRecordData(syncPriceRecordDataRequest);
            log.info("MnSapIf001V2Service_externalSyncPrice_save_price_record_request={},result={}", syncPriceRecordDataRequest.toString(),JSONObject.toJSONString(syncPriceRecordDataResponse));

            if (Objects.isNull(syncPriceRecordDataResponse) || !syncPriceRecordDataResponse.getSuccess()) {
                mnSapCommSDO.setStatus(MnSAPStatus.ERROR.getCode());
                mnSapCommSDO.setMessage("同步失败");
            }else{
                successNum++;
            }
            //同步后的结果封装
            CallApiRecordResponse callAPISyncPriceRecordDataResponse = new CallApiRecordResponse();
            callAPISyncPriceRecordDataResponse.setOperateResult(syncPriceRecordDataResponse.getSuccess());
            callAPISyncPriceRecordDataResponse.setRequestBody(JSONObject.toJSONString(syncPriceRecordDataRequest));
            callAPISyncPriceRecordDataResponse.setTableName("ofs_price_record");
            callAPISyncPriceRecordDataResponse.setMessage(syncPriceRecordDataResponse.getMessage());
            callApiRecordResponses.add(callAPISyncPriceRecordDataResponse);
        }

        log.info("MnSapIf001V2Service_externalSyncPrice_receive_data_tableName=knop,sync_success_size={}",successNum);

        return count;
    }

    private CompletableFuture<Long> saveKnopPriceSyncTempTable(Map<String, Object> data,String messageId,String sendTime,String sender){
        return CompletableFuture.supplyAsync(() -> {
            BatchDeleteRequest BatchDeleteRequest = new BatchDeleteRequest();
            BatchDeleteRequest.setMessageId(messageId);
            BatchDeleteResponse batchDeleteResponse = priceSyncTempService.batchDelete(BatchDeleteRequest);
            log.info("MnSapIf001V2Service_saveKnopPriceSyncTempTable_batchDeleteResponse",batchDeleteResponse);

            Long count = 0L;
            Object konp = data.get("KONP");
            if (Objects.isNull(konp)) {
                return count;
            }
            List<PriceSyncTemp> knopList = JSONObject.parseArray(JSONObject.toJSONString(konp), PriceSyncTemp.class);
            log.info("MnSapIf001V2Service_saveKnopPriceSyncTempTable_receive_data_tableName=knop,receive_size={}",knopList.size());
            if (CollectionUtils.isEmpty(knopList)) {
                return count;
            }
            count += knopList.size();

            Long finalCount = count;
            knopList.forEach(x->{
                x.setTableKey("KONP");
                x.setMessageId(messageId);
                x.setSendTime(sendTime);
                x.setSender(sender);
                x.setStatus("0");
                x.setRequestTotal(String.valueOf(finalCount));
            });

            int pageSize = 1000;
            int totalPage = (knopList.size() + pageSize - 1)/pageSize;

            for (int pageNo = 0; pageNo < totalPage; pageNo++) {
                int start = pageNo * pageSize;
                int end = pageNo+1 == totalPage ? knopList.size(): start + pageSize;
                BatchCreateRequest request = new BatchCreateRequest();
                request.setCreateList(knopList.subList(start, end));
                Object response = priceSyncTempService.batchCreate(request);
                log.info("MnSapIf001V2Service_saveKnopPriceSyncTempTable_batchCreate_result={},isEndPage={}",JSONObject.toJSON(response),pageNo+1 == totalPage);
            }

            return count;
        }, scItemsQueryTaskExecutor);
    }

    private CompletableFuture<Long> saveDimensionPriceSyncTempTable(Map<String, Object> data,String messageId,String sendTime,String sender){
        return CompletableFuture.supplyAsync(() -> {
            com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.BatchDeleteRequest batchDeleteRequest = new com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.BatchDeleteRequest();
            batchDeleteRequest.setMessageId(messageId);
            com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.BatchDeleteResponse batchDeleteResponse = priceSyncTempDimensionService.batchDelete(batchDeleteRequest);
            log.info("MnSapIf001V2Service_saveDimensionPriceSyncTempTable_batchDeleteResponse={}",batchDeleteResponse);

            long count = 0;
            List<String> tableDataList = data.keySet().stream().filter(key -> !StringUtils.equals(key, "KONP")).collect(Collectors.toList());
            log.info("MnSapIf001V2Service_saveDimensionPriceSyncTempTable_receive_data_tableDataList={}",tableDataList);

            for (String tableName : tableDataList){
                List<Object> table = (List<Object>) data.get(tableName);
                count += table.size();
            }

            for (String tableName : tableDataList) {
                List<Object> table = (List<Object>) data.get(tableName);
                log.info("MnSapIf001V2Service_saveDimensionPriceSyncTempTable_receive_data_tableName_forEach_tableName={},receive_size={}",tableName,table.size());
                if (CollectionUtils.isEmpty(table)) {
                    continue;
                }
                List<PriceSyncTempDimension> dimensionList = JSONObject.parseArray(JSONObject.toJSONString(table), PriceSyncTempDimension.class);
                if (CollectionUtils.isEmpty(dimensionList)) {
                    continue;
                }

                long finalCount = count;
                dimensionList.forEach(x->{
                    x.setTableKey(tableName);
                    x.setMessageId(messageId);
                    x.setSendTime(sendTime);
                    x.setSender(sender);
                    x.setStatus("0");
                    x.setRequestTotal(String.valueOf(finalCount));
                });

                int pageSize = 1000;
                int totalPage = (dimensionList.size() + pageSize - 1)/pageSize;

                for (int pageNo = 0; pageNo < totalPage; pageNo++) {
                    int start = pageNo * pageSize;
                    int end = pageNo+1 == totalPage ? dimensionList.size(): start + pageSize;
                    com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.BatchCreateRequest request = new com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.BatchCreateRequest();
                    request.setCreateList(dimensionList.subList(start, end));
                    Object response = priceSyncTempDimensionService.batchCreate(request);
                    log.info("MnSapIf001V2Service_saveDimensionPriceSyncTempTable_batchCreate_result={},isEndPage={}",JSONObject.toJSON(response),pageNo+1 == totalPage);
                }
            }
            return count;
        }, scItemsQueryTaskExecutor);
    }

    @Override
    public Result<Void> doHandlerTempTable(DoHandlerTempTableRequest doHandlerTempTableRequest) {
        PageRecordUtil pageUtil = new PageRecordUtil();
        pageUtil.simpleResetPageParam();
        //messageId分组查询，获取所有未处理的请求id
        for (int i = 1; i <= pageUtil.getPage(); i++) {
            com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.GroupByMessageIdRequest queryRequest = new com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.GroupByMessageIdRequest();
            queryRequest.setLimit(pageUtil.getPageSize());
            queryRequest.setStart(pageUtil.getStart());
            queryRequest.setMessageId(doHandlerTempTableRequest.getMessageId());
            queryRequest.setOrgSalesOrganizationCode(doHandlerTempTableRequest.getOrgSalesOrganizationCode());
            Result<List<PriceSyncTemp>> messageIdResult = priceSyncTempDimensionService.groupByMessageId(queryRequest);
            log.info("MnSapIf001V2Service_doHandlerTempTable_groupByMessageId_request={},result={},current_page={}",JSONObject.toJSONString(queryRequest),JSONObject.toJSONString(messageIdResult.getResult()),i);
            if(i == 1){
                pageUtil.setTotalPages(messageIdResult.getTotal());
                log.info("MnSapIf001V2Service_doHandlerTempTable_groupByMessageId_totalMessageId_size={},totalpage={}",messageIdResult.getTotal(),pageUtil.getPage());
            }

            //根据查出的按sendTime排序的，messageId依次执行
            dealMessageIdRequest(messageIdResult.getResult());
            pageUtil.turnPage();
        }

        if(doHandlerTempTableRequest.getDealRecordTempTable() != null && doHandlerTempTableRequest.getDealRecordTempTable()){
            PageRecordUtil pageUtil2 = new PageRecordUtil();
            pageUtil2.simpleResetPageParam();
            for (int i = 1; i <= pageUtil2.getPage(); i++) {
                GroupByMessageIdRequest queryRequest2 = new GroupByMessageIdRequest();
                queryRequest2.setLimit(pageUtil2.getPageSize());
                queryRequest2.setStart(pageUtil2.getStart());
                Result<List<PriceSyncTemp>> messageIdResult = priceSyncTempService.groupByMessageId(queryRequest2);
                log.info("MnSapIf001V2Service_doHandlerTempTable_groupByMessageId2_current_page={},messageIds={}",i,JSONObject.toJSONString(messageIdResult.getResult()));
                if(i == 1){
                    pageUtil2.setTotalPages(messageIdResult.getTotal());
                    log.info("MnSapIf001V2Service_doHandlerTempTable_groupByMessageId2_totalMessageId_size={},totalpage={}",messageIdResult.getTotal(),pageUtil2.getPage());
                }
                //根据查出的按sendTime排序的，messageId依次执行
                dealMessageIdRequest(messageIdResult.getResult());
                pageUtil2.turnPage();
            }
        }
        return Result.success(null);
    }

    private void dealMessageIdRequest(List<PriceSyncTemp> list){
        //根据查出的按sendTime排序的，messageId依次执行。  保证messageId处理的事务性
        for (PriceSyncTemp sdo :list) {
            try {
                String messageId = sdo.getMessageId();
                //组装messageHeader
                Map<String, Object> messageHeader = Maps.newHashMap();
                messageHeader.put(MESSAGE_ID,messageId);
                messageHeader.put(DATA_SOURCE,sdo.getSender());
                //组装data
                Map<String, Object> data = Maps.newHashMap();
                List<PriceSyncTemp> knopList = getPriceSyncTempList(messageId);
                data.put("KONP",knopList);
                List<PriceSyncTempDimension> dimensionList = getPriceSyncTempDimensionList(messageId);
                Map<String,List<PriceSyncTempDimension>> dimensionTableMap = Nullable.stream(dimensionList).collect(Collectors.groupingBy(x->x.getTableKey()));
                data.putAll(dimensionTableMap);

                Long querySize = (Nullable.stream(dimensionTableMap.values()).mapToLong(x->x.size()).sum()) + Long.valueOf(knopList.size());
                Long requestSize = 0L;
                if(CollectionUtils.isNotEmpty(knopList)){
                    requestSize += Long.valueOf(knopList.get(0).getRequestTotal());
                }
                if(CollectionUtils.isNotEmpty(dimensionList)){
                    requestSize += Long.valueOf(dimensionList.get(0).getRequestTotal());
                }
                //保证拉取的数据与请求的数据一致，才处理。
                if(!querySize.equals(requestSize)){
                    log.info("MnSapIf001V2Service_doHandlerTempTable_fetchList_size_error_messageId={}",messageId);
                    continue;
                }

                MnSapCommSDO mnSapCommSDO = new MnSapCommSDO();
                //操作过的表结果记录
                List<CallApiRecordResponse> callApiRecordResponses = Lists.newArrayList();

                long konpCount = extractedPriceRecord(data, mnSapCommSDO, callApiRecordResponses,messageHeader);

                Map<String, PriceDimensionResponse> dimensionMap = queryPriceDimensionResponseMap(data);

                long tableCount = savePricePropertyConfig(data, mnSapCommSDO, callApiRecordResponses, dimensionMap,messageHeader);
                log.info("MnSapIf001V2Service_doHandlerTempTable_receive_tableCounts_total_size={}", tableCount);
                log.info("MnSapIf001V2Service_doHandlerTempTable_sync_result={}", JSONObject.toJSONString(callApiRecordResponses));

                //记录收到的数据条数
                mnSapCommSDO.setDref1(konpCount + tableCount);
                //没有失败记录时表示同步数据成功
                if (StringUtils.isBlank(mnSapCommSDO.getStatus()) && StringUtils.isBlank(mnSapCommSDO.getMessage())) {
                    mnSapCommSDO.setStatus(MnSAPStatus.SUCCESS.getCode());
                    mnSapCommSDO.setMessage("同步成功");
                }

                MnSapCommResponse mnSapCommResponse = new MnSapCommResponse();
                mnSapCommResponse.setInterfacePath((String) messageHeader.get("interface"));
                mnSapCommResponse.setMessageId((String) messageHeader.get("messageId"));
                List<MnSapCommSDO> mnSapCommSDOS = Lists.newArrayList(mnSapCommSDO);
                mnSapCommResponse.setData(mnSapCommSDOS);
                log.info("MnSapIf001V2Service_doHandlerTempTable_sync_messageId_result={}", JSONObject.toJSONString(mnSapCommResponse));

                Long dealSize = konpCount + tableCount;
                if(MnSAPStatus.SUCCESS.getCode().equals(mnSapCommSDO.getStatus())
                        && querySize.equals(dealSize)){
                    UpdateRequest updateRequest = new UpdateRequest();
                    updateRequest.setMessageId(messageId);
                    updateRequest.setStatus("1");
                    UpdateResponse response = priceSyncTempService.update(updateRequest);
                    if(!Long.valueOf(response.getCount()).equals(konpCount)){
                        log.info("MnSapIf001V2Service_doHandlerTempTable_updatePriceSyncTempServiceList_size_error_messageId={},updateSize={}",messageId,response.getCount());
                    }
                    com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.UpdateRequest updateRequest2 = new com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.UpdateRequest();
                    updateRequest2.setMessageId(messageId);
                    updateRequest2.setStatus("1");
                    com.epoch.app.otsapplicationadaptersdk.pricesynctempdimension.dto.UpdateResponse response2 = priceSyncTempDimensionService.update(updateRequest2);
                    if(!Long.valueOf(response2.getCount()).equals(tableCount)){
                        log.info("MnSapIf001V2Service_doHandlerTempTable_updatePriceSyncTempDimensionServiceList_size_error_messageId={},updateSize={}",messageId,response2.getCount());
                    }
                    //如果处理的数据与请求的数量不符，不改处理状态了。。。
                }else{
                    log.info("MnSapIf001V2Service_doHandlerTempTable_dealList_size_error_messageId={}",messageId);
                }
            } catch (Exception e) {
                log.info("MnSapIf001V2Service_doHandlerTempTable_exception_messageId={},error={}",sdo.getMessageId(),e.getMessage());
            }
        }
    }

    private List<PriceSyncTemp> getPriceSyncTempList(String messageId){
        List<Future<List<PriceSyncTemp>>> futureList = Lists.newArrayList();
        int pageSize = 1000;
        PriceSyncTempLoadListVO queryRequest = new PriceSyncTempLoadListVO();
        queryRequest.setStatus("0");
        queryRequest.setMessageId(messageId);
        queryRequest.setStart(0);
        queryRequest.setLimit(1);
        Result<List<PriceSyncTemp>> knopResponse = priceSyncTempService.loadListForPage(queryRequest);
        int total = knopResponse.getTotal();
        int totalPage = (total + pageSize - 1)/pageSize;

        for (int pageNo = 0; pageNo < totalPage; pageNo++) {
            int start = pageNo * pageSize;
            CompletableFuture<List<PriceSyncTemp>> future = CompletableFuture.supplyAsync(() -> {
                Result<List<PriceSyncTemp>> planPageResult = null;
                try {
                    PriceSyncTempLoadListVO queryRequest2 = new PriceSyncTempLoadListVO();
                    queryRequest2.setStatus("0");
                    queryRequest2.setMessageId(messageId);
                    queryRequest2.setStart(start);
                    queryRequest2.setLimit(pageSize);
                    planPageResult = priceSyncTempService.loadListForPage(queryRequest2);
                    if (planPageResult==null || CollectionUtils.isEmpty(planPageResult.getResult())){
                        return Lists.newArrayList();
                    }
                } catch (Exception e) {
                    log.error("MnSapIf001V2Service_doHandlerTempTable_getPriceSyncTempList_error_messageId={},error={}", messageId, e.getMessage());
                    return Lists.newArrayList();
                }
                return planPageResult.getResult();
            }, scItemsQueryTaskExecutor);
            futureList.add(future);
        }

        List<PriceSyncTemp> result = new ArrayList<>();
        try {
            for (Future<List<PriceSyncTemp>> future : futureList){
                List<PriceSyncTemp> subList = future.get();
                result.addAll(subList);
            }
        }catch (Exception e){
            log.error("MnSapIf001V2Service_doHandlerTempTable_getPriceSyncTempList_future_get_error={}", e.getMessage());
        }
        return result;
    }

    private List<PriceSyncTempDimension> getPriceSyncTempDimensionList(String messageId){
        List<Future<List<PriceSyncTempDimension>>> futureList = Lists.newArrayList();
        int pageSize = 1000;
        PriceSyncTempDimensionLoadListVO queryRequest = new PriceSyncTempDimensionLoadListVO();
        queryRequest.setStatus("0");
        queryRequest.setMessageId(messageId);
        queryRequest.setStart(0);
        queryRequest.setLimit(1);
        Result<List<PriceSyncTempDimension>> knopResponse = priceSyncTempDimensionService.loadListForPage(queryRequest);
        int total = knopResponse.getTotal();
        int totalPage = (total + pageSize - 1)/pageSize;

        for (int pageNo = 0; pageNo < totalPage; pageNo++) {
            int start = pageNo * pageSize;
            CompletableFuture<List<PriceSyncTempDimension>> future = CompletableFuture.supplyAsync(() -> {
                Result<List<PriceSyncTempDimension>> planPageResult = null;
                try {
                    PriceSyncTempDimensionLoadListVO queryRequest2 = new PriceSyncTempDimensionLoadListVO();
                    queryRequest2.setStatus("0");
                    queryRequest2.setMessageId(messageId);
                    queryRequest2.setStart(start);
                    queryRequest2.setLimit(pageSize);
                    planPageResult = priceSyncTempDimensionService.loadListForPage(queryRequest2);
                    if (planPageResult==null || CollectionUtils.isEmpty(planPageResult.getResult())){
                        return Lists.newArrayList();
                    }
                } catch (Exception e) {
                    log.error("MnSapIf001V2Service_doHandlerTempTable_getPriceSyncTempDimensionList_error_messageId={},error={}", messageId, e.getMessage());
                    return Lists.newArrayList();
                }
                return planPageResult.getResult();
            }, scItemsQueryTaskExecutor);
            futureList.add(future);
        }

        List<PriceSyncTempDimension> result = new ArrayList<>();
        try {
            for (Future<List<PriceSyncTempDimension>> future : futureList){
                List<PriceSyncTempDimension> subList = future.get();
                result.addAll(subList);
            }
        }catch (Exception e){
            log.error("MnSapIf001V2Service_doHandlerTempTable_getPriceSyncTempDimensionList_future_get_error={}", e.getMessage());
        }

        //Map<String,List<PriceSyncTempDimension>> map = Nullable.stream(result).collect(Collectors.groupingBy(x->x.getTableKey()));
        return result;
    }

}
